package com.syher.seconds.kill.kafka.core.kafka;

import com.alibaba.fastjson.JSONObject;
import com.syher.seconds.kill.kafka.core.SpringBeanFactory;
import com.syher.seconds.kill.pojo.Stock;
import com.syher.seconds.kill.service.IOrderService;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.RejectedExecutionException;

public class ConsumerTask implements Runnable {
    private static Logger LOGGER = LoggerFactory.getLogger(ConsumerTask.class);

    private final KafkaConsumer<String, String> consumer;
    private final IOrderService orderService;

    public ConsumerTask(KafkaConsumer<String, String> consumer) {
        this.consumer = consumer;
        orderService = SpringBeanFactory.getBean(IOrderService.class);
    }

    @Override
    public void run() {
        // 使用200ms作为获取超时时间
        ConsumerRecords<String, String> records = consumer.poll(200);

        for (ConsumerRecord<String, String> record : records) {
            // 简单地打印消息
            LOGGER.info("===" + record.value() + " consumed " + record.partition() +
                    " message with offset: " + record.offset());

            dealMessage(record.value());
        }
    }

    private void dealMessage(String value) {
        try {

            Stock stock = JSONObject.parseObject(value, Stock.class);
            LOGGER.info("consumer stock={}", JSONObject.toJSONString(stock));

            //创建订单
            orderService.createOptimisticOrderUseRedisAndKafka(stock.getId());

        } catch (RejectedExecutionException e) {
            LOGGER.error("rejected message = " + value);
        } catch (Exception e) {
            LOGGER.error("unknown exception", e);
        }

    }
}
